/*
 * Copyright 2019 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */
package io.netty.example.fep.netty.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.example.fep.netty.common.FepProtocolDecoder;
import io.netty.example.fep.netty.common.FepProtocolEncoder;
import io.netty.example.fep.netty.common.FepProtocolMsg;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Description：
 * netty 服务端
 *
 * @author : chenjie03@imdada.cn
 * Date: 2020/5/26
 * Time: 13:50
 */
public final class NettyServer {

    private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);

    /**
     * netty 服务端 端口号
     */
    private int port;

    /**
     * orgCode - channelId 映射表
     */
    private Map<String, Set<String>> clientMap = new ConcurrentHashMap<String, Set<String>>();

    /**
     * channelId - channel 映射表
     */
    private Map<String, Channel> channelMap = new ConcurrentHashMap<String, Channel>();


    public NettyServer(int port, ServerHandler handler) {
        this.port = port;
        init(handler);
    }

    /**
     * 初始化服务端
     */
    private void init(ServerHandler handler) {
        try {
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup(10);
            ServerBootstrap bootstrap = new ServerBootstrap();
            final NettyServer server = this;
            final ServerHandler myHandler = handler;
            bootstrap.group(bossGroup, workerGroup);
            bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
            bootstrap.channel(NioServerSocketChannel.class);
            bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    ch.pipeline().addLast("encoder", new FepProtocolEncoder());
                    ch.pipeline().addLast("decoder", new FepProtocolDecoder());
                    ch.pipeline().addLast("heartBeatHandler", new IdleStateHandler(45, 0, 0, TimeUnit.SECONDS));
                    ch.pipeline().addLast("handler", new ServerMainHandler(server, myHandler));
                }
            });
            bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
            bootstrap.bind(port).sync();
            logger.info("server 启动完成");
        } catch (InterruptedException e) {
            logger.error("server 启动失败，e=", e);
        }
    }

    /**
     * 服务端发送自定义消息
     *
     * @param orgCode 中台商家编码
     * @param type    消息类型
     * @param msg     消息体
     * @throws Exception 中断异常
     * @see FepProtocolMsg.Constants 消息类型
     */
    public boolean send(String orgCode, byte type, String msg) {
        try {
            logger.info("server => client(clientId={}) type:{}, msg:{}", orgCode, type, msg);
            //组装消息体
            FepProtocolMsg protocol = new FepProtocolMsg(type, msg);
            //获取channelId
            Set<String> channelIdList = clientMap.get(orgCode);
            if (channelIdList == null || channelIdList.isEmpty()) {
                return false;
            }
            //向所有channel广播
            for (String channelId : channelIdList) {
                Channel channel = channelMap.get(channelId);
                if (channel != null && channel.isActive()) {
                    channel.writeAndFlush(protocol).sync();
                }
            }
            return true;
        } catch (Exception e) {
            logger.error("server => client(clientId=" + orgCode + ") 发送消息失败，e=", e);
        }
        return false;
    }


    /**
     * 缓存所有管道
     *
     * @param channelId 管道ID
     * @param channel   客户端channel
     */
    public void putChannel(String channelId, Channel channel) {
        this.channelMap.put(channelId, channel);
    }

    /**
     * 缓存某一前置机所有管道ID
     *
     * @param orgCode   中台商家编码
     * @param channelId 管道ID
     */
    public void putClient(String orgCode, String channelId) {
        Set<String> channelIdList = this.clientMap.get(orgCode);
        if (channelIdList == null) {
            channelIdList = new HashSet<String>();
        }
        channelIdList.add(channelId);
        this.clientMap.put(orgCode, channelIdList);
    }

    /**
     * 移除失连管道ID
     *
     * @param channelId 管道ID
     */
    public void removeChannel(final String channelId) {
        Iterator<Map.Entry<String, Channel>> iterator = this.channelMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, Channel> next = iterator.next();
            if (next.getValue().id().asLongText().equals(channelId)) {
                this.clientMap.remove(next.getKey());
            }
        }
    }

}
